 1.Kafka高级特性解析之稳定性中一致性保证
   
   一、概念
   1).水位标记
   水位或水印（watermark）一词，表示位置信息，即位移（offset）。Kafka源码中使用的名字是高水位，HW
（high watermark）。
   2).副本角色
   Kafka分区使用多个副本(replica)提供高可用。
   3).LEO和HW
   每个分区副本对象都有两个重要的属性：LEO和HW。
      LEO：即日志末端位移(log end offset)，记录了该副本日志中下一条消息的位移值。如果LEO=10,那么表示
该副本保存了10条消息，位移值范围是[0, 9]。另外，Leader LEO和Follower LEO的更新是有区别的。
      HW：即上面提到的水位值。对于同一个副本对象而言，其HW值不会大于LEO值。小于等于HW值的所有消息都被
认为是“已备份”的（replicated）。Leader副本和Follower副本的HW更新不同。 
   上图中，HW值是7，表示位移是0~7 的所有消息都已经处于“已提交状态”（committed），而LEO值是14，8~13
的消息就是未完全备份（fully replicated）——为什么没有14？LEO指向的是下一条消息到来时的位移。
   消费者无法消费分区下Leader副本中位移大于分区HW的消息。
   二、Follower副本何时更新LEO
   Follower副本不停地向Leader副本所在的broker发送FETCH请求，一旦获取消息后写入自己的日志中进行备份。
那么Follower副本的LEO是何时更新的呢？⾸先我必须⾔明，Kafka有两套Follower副本LEO：
   1).一套LEO保存在Follower副本所在Broker的副本管理机中；
   2).另一套LEO保存在Leader副本所在Broker的副本管理机中。Leader副本机器上保存了所有的follower副本的
LEO。
   Kafka使用前者帮助Follower副本更新其HW值；利用后者帮助Leader副本更新其HW。
   1).Follower副本的本地LEO何时更新？
   Follower副本的LEO值就是日志的LEO值，每当新写入一条消息，LEO值就会被更新。当Follower发送FETCH请求后
，Leader将数据返回给Follower，此时Follower开始Log写数据，从而自动更新LEO值。
   2).Leader端Follower的LEO何时更新？
   Leader端的Follower的LEO更新发生在Leader在处理Follower FETCH请求时。一旦Leader接收到Follower发送的
FETCH请求，它先从Log中读取相应的数据，给Follower返回数据前，先更新Follower的LEO。
   三、Follower副本何时更新HW
   Follower更新HW发生在其更新LEO之后，一旦Follower向Log写完数据，尝试更新自己的HW值。
   比较当前LEO值与FETCH响应中Leader的HW值，取两者的小者作为新的HW值。
   即：如果Follower的LEO大于Leader的HW，Follower HW值不会大于Leader的HW值。
   四、Leader副本何时更新LEO
   和Follower更新LEO相同，Leader写Log时自动更新自己的LEO值。
   五、Leader副本何时更新HW值
   Leader的HW值就是分区HW值，直接影响分区数据对消费者的可见性 。
   Leader会尝试去更新分区HW的四种情况：
   1).Follower副本成为Leader副本时：Kafka会尝试去更新分区HW。
   2).Broker崩溃导致副本被踢出ISR时：检查下分区HW值是否需要更新是有必要的。
   3).生产者向Leader副本写消息时：因为写入消息会更新Leader的LEO，有必要检查HW值是否需要更新
   4). Leader处理Follower FETCH请求时：首先从Log读取数据，之后尝试更新分区HW值
   结论：
   当Kafka broker都正常工作时，分区HW值的更新时机有两个：
     1). Leader处理PRODUCE请求时
     2). Leader处理FETCH请求时。
   Leader如何更新自己的HW值？Leader broker上保存了一套Follower副本的LEO以及自己的LEO。当尝试确定分
区HW时，它会选出所有满足条件的副本，比较它们的LEO(包括Leader的LEO)，并选择最小的LEO值作为HW值。
   需要满足的条件，（二选一）：
      1). 处于ISR中
      2). 副本LEO落后于Leader LEO的时长不大于replica.lag.time.max.ms 参数值(默认是10s)
   如果Kafka只判断第一个条件的话，确定分区HW值时就不会考虑这些未在ISR中的副本，但这些副本已经具备了
“立刻进入ISR”的资格，因此就可能出现分区HW值越过ISR中副本LEO的情况——不允许。因为分区HW定义就是ISR中
所有副本LEO的最小值。
   六、HW和LEO正常更新案例
   我们假设有一个topic，单分区，副本因⼦是2，即一个Leader副本和一个Follower副本。我们看下当producer发
送一条消息时，broker端的副本到底会发⽣什么事情以及分区HW是如何被更新的。
   1).初始状态
   初始时Leader和Follower的HW和LEO都是0(严格来说源代码会初始化LEO为-1,不过这不影响之后的讨论)。
Leader中的Remote LEO指的就是Leader端保存的Follower LEO，也被初始化成0。此时，生产者没有发送任何消息
给Leader，而Follower已经开始不断地给Leader发送FETCH请求了，但因为没有数据因此什么都不会发生。值得一提
的是，Follower发送过来的FETCH请求因为无数据而暂时会被寄存到Leader端的purgatory中，待500ms
( replica.fetch.wait.max.ms 参数)超时后会强制完成。倘若在寄存期间生产者发来数据，则Kafka会自动唤醒该
FETCH请求，让Leader继续处理。
   2).Follower发送FETCH请求在Leader处理完PRODUCE请求之后
   producer给该topic分区发送了一条消息
   此时的状态如下图所示：
   如上图所示，Leader接收到PRODUCE请求主要做两件事情：
   (1).把消息写入Log，同时自动更新Leader自己的LEO
   (2).尝试更新Leader HW值。假设此时Follower尚未发送FETCH请求，Leader端保存的Remote LEO依然是0，
因此Leader会比较它自己的LEO值和Remote LEO值，发现最小值是0，与当前HW值相同，故不会更新分区
HW值（仍为0）
   PRODUCE请求处理完成后各值如下，Leader端的HW值依然是0，而LEO是1，Remote LEO也是0。
   属性         阶段            旧值 新值  备注
   Leader LEO   PRODUCE处理完成  0    1    写入了一条数据
   Remote LEO   PRODUCE处理完成  0    0    还未Fetch
   Leader HW    PRODUCE处理完成  0    0    min(LeaderLEO=1, RemoteLEO=0)=0
   Follower LEO PRODUCE处理完成  0    0    还未Fetch
   Follower HW  PRODUCE处理完成  0    0    min(LeaderHW=0, FollowerLEO=0)=0
   假设此时follower发送了FETCH请求，则状态变更如下：
   本例中当follower发送FETCH请求时，Leader端的处理依次是：
   (1).读取Log数据
   (2).更新remote LEO = 0（为什么是0？ 因为此时Follower还没有写入这条消息。Leader如何确认Follower还未
写入呢？这是通过Follower发来的FETCH请求中的Fetch offset来确定的）
   (3).尝试更新分区HW：此时Leader LEO = 1,Remote LEO = 0,故分区HW值= min(Leader LEO, Follower
Remote LEO) = 0
   (4).把数据和当前分区HW值（依然是0）发送给Follower副本
   而Follower副本接收到FETCH Response后依次执行下列操作：
   (1). 写入本地Log，同时更新Follower自己管理的 LEO为1
   (2). 更新Follower HW：比较本地LEO和 FETCH Response 中的当前Leader HW值，取较小者，Follower HW =0
   此时，第一轮FETCH RPC结束，我们会发现虽然Leader和Follower都已经在Log中保存了这条消息，但分区HW值
尚未被更新，仍为0。
   属性          阶段                         旧值 新值 备注
   Leader LEO    第二次Follower FETCH处理完成  1    1   未写入新数据
   Remote LEO    第二次Follower FETCH处理完成  0    1   第2次fetch中offset为1
   Leader HW     第二次Follower FETCH处理完成  0    1   min(RemoteLEO,LeaderLEO)=1
   Follower LEO  第二次Follower FETCH处理完成  1    1   未写入新数据
   Follower HW   第二次Follower FETCH处理完成  0    1   第2次fetch resp中的LeaderHW和本地FollowerLEO都是1
   此时消息已经成功地被复制到Leader和Follower的Log中且分区HW是1，表明消费者能够消费offset =0 的消息。
   3).FETCH请求保存在purgatory中，PRODUCE请求到来。
   当Leader无法立即满足FECTH返回要求的时候(比如没有数据)，那么该FETCH请求被暂存到Leader端的purgatory
中（炼狱），待时机成熟尝试再次处理。Kafka不会无限期缓存，默认有个超时时间（500ms），一旦超时时间已过，
则这个请求会被强制完成。当寄存期间还没超时，生产者发送PRODUCE请求从而使之满足了条件以致被唤醒。此时，
Leader端处理流程如下：
   (1). Leader写Log（自动更新Leader LEO）
   (2). 尝试唤醒在purgatory中寄存的FETCH请求
   (3). 尝试更新分区HW
   七、HW和LEO异常案例
   Kafka使用HW值来决定副本备份的进度，而HW值的更新通常需要额外一轮FETCH RPC才能完成。但这种设计是
有问题的，可能引起的问题包括：
   (1).备份数据丢失
   (2).备份数据不一致
    1).数据丢失
   使用HW值来确定备份进度时其值的更新是在下一轮RPC中完成的。如果Follower副本在标记上方的的第一步与第
二步之间发生崩溃，那么就有可能造成数据的丢失。
   上图中有两个副本：A和B。开始状态是A是Leader。
   假设生产者min.insync.replicas 为1，那么当生产者发送两条消息给A后，A写入Log，此时Kafka会通知生产者
这两条消息写入成功。
   代  属性         阶段                            旧值 新值   备注
   1   Leader LEO   PRODUCE和Follower FETCH处理完成  0    1     写入了一条数据
   1   Remote LEO   PRODUCE和Follower FETCH处理完成  0    0     第一次fetch中offset为0
   1   Leader HW    PRODUCE和Follower FETCH处理完成  0    0     min(LeaderLEO=1,FollowerLEO=0)=0
   1   Follower LEO PRODUCE和Follower FETCH处理完成  0    1     同步了一条数据
   1   Follower HW  PRODUCE和Follower FETCH处理完成  0    0     min(LeaderHW=0, FollowerLEO=1)=0
   2   Leader LEO   第二次Follower FETCH处理完成     1    2     写入了第二条数据
   2   Remote LEO   第二次Follower FETCH处理完成     0    1     第2次fetch中offset为1
   2   Leader HW    第二次Follower FETCH处理完成     0    1     min(RemoteLEO=1,LeaderLEO=2)=1
   2   Follower LEO 第二次Follower FETCH处理完成     1    2     写入了第二条数据
   2   Follower HW  第二次Follower FETCH处理完成     0    1     min(LeaderHW=1,FollowerLEO=2)=1
   3   Leader LEO   第三次Follower FETCH处理完成     2    2     未写入新数据
   3   Remote LEO   第三次Follower FETCH处理完成     1    2     第3次fetch中offset为2
   3   Leader HW    第三次Follower FETCH处理完成     1    2     min(RemoteLEO=2,LeaderLEO)=2
   3   Follower LEO 第三次Follower FETCH处理完成     2    2     未写入新数据
   3   Follower HW  第三次Follower FETCH处理完成     1    2     第3次fetch resp中的LeaderHW和本地FollowerLEO都是2
   但是在broker端，Leader和Follower的Log虽都写入了2条消息且分区HW已经被更新到2，但Follower HW尚未被
更新还是1，也就是上面标记的第二步尚未执行，表中最后一条未执行。
   倘若此时副本B所在的broker宕机，那么重启后B会⾃动把LEO调整到之前的HW值1，故副本B会做⽇志截断(log
truncation)，将offset =1 的那条消息从log中删除，并调整LEO =1. 此时follower副本底层log中就只有一条消息，即
offset = 0 的消息！
   B重启之后需要给A发FETCH请求，但若A所在broker机器在此时宕机，那么Kafka会令B成为新的Leader，而当A
重启回来后也会执行日志截断，将HW调整回1。这样，offset=1 的消息就从两个副本的log中被删除，也就是说这条已
经被生产者认为发送成功的数据丢失。
   丢失数据的前提是min.insync.replicas=1 时，一旦消息被写入Leader端Log即被认为是committed 。延迟一
轮FETCH RPC 更新HW值的设计使follower HW值是异步延迟更新，若在这个过程中Leader发生变更，那么成为新
Leader的Follower的HW值就有可能是过期的，导致生产者本是成功提交的消息被删除。
   2).Leader和Follower数据离散
   除了可能造成的数据丢失以外，该设计还会造成Leader的Log和Follower的Log数据不一致。
   如Leader端记录序列：m1,m2,m3,m4,m5,…；Follower端序列可能是m1,m3,m4,m5,…。
   看图：
   假设：A是Leader，A的Log写入了2条消息，但B的Log只写了1条消息。分区HW更新到2，但B的HW还是1，同时
生产者min.insync.replicas 仍然为1。
   假设A和B所在Broker同时宕机，B先重启回来，因此B成为Leader，分区HW =1. 假设此时生产者发送了第3条
消息(红色表示)给B，于是B的log中offset = 1 的消息变成了红框表示的消息，同时分区HW更新到2（A还没有回来，就
B一个副本，故可以直接更新HW而不用理会A）之后A重启回来，需要执行日志截断，但发现此时分区HW=2 而A之前
的HW值也是2，故不做任何调整。此后A和B将以这种状态继续正常工作。
   八、Leader Epoch使用
   0. Kafka解决方案
   造成上述两个问题的根本原因在于
   (1). HW值被用于衡量副本备份的成功与否。
   (2). 在出现失败重启时作为日志截断的依据。
   但HW值的更新是异步延迟的，特别是需要额外的FETCH请求处理流程才能更新，故这中间发生的任何崩溃都可能
导致HW值的过期。
   Kafka从0.11 引入了leader epoch 来取代HW值。Leader端使用内存保存Leader的epoch信息，即使出现上面的
两个场景也能规避这些问题。
   所谓Leader epoch实际上是一对值：<epoch, offset>：
   (1). epoch表示Leader的版本号，从0开始，Leader变更过1次，epoch+1
   (2). offset对应于该epoch版本的Leader写⼊第⼀条消息的offset。因此假设有两对值：
   <0, 0>
   <1, 120>
   则表示第一个Leader从位移0开始写⼊消息；共写了120条[0, 119]；而第二个Leader版本号是1，从位移120处开
始写入消息。
   (1).Leader broker中会保存这样的一个缓存，并定期地写入到一个checkpoint 文件中。
   (2).当Leader写Log时它会尝试更新整个缓存：如果这个Leader首次写消息，则会在缓存中增加一个条目；否则
就不做更新。
   (3).每次副本变为Leader时会查询这部分缓存，获取出对应Leader版本的位移，则不会发生数据不一致和丢失的
情况。
   1).规避数据丢失
   只需要知道每个副本都引入了新的状态来保存⾃⼰当leader时开始写入的第一条消息的offset以及leader版本。这
样在恢复的时候完全使用这些信息而非HW来判断是否需要截断日志。
   2).规避数据不一致
   依靠Leader epoch的信息可以有效地规避数据不一致的问题。
  
   